package hust.rpc.handler;

import hust.rpc.ConsumerBootstrap;
import hust.rpc.constants.Constants;
import hust.rpc.invoker.RemoteInvoker;
import hust.rpc.pojo.Host;
import hust.rpc.pojo.RpcResponse;
import hust.rpc.proxy.Proxy;
import hust.rpc.service.ConsumerService;
import hust.rpc.utils.SingletonUtils;
import hust.rpc.utils.ThreadUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;

import java.util.*;

/**
 * 消费者 <-> 注册中心处理器
 */
@Slf4j
public class RegistryHandler extends RpcInvokeHandler {

    /**
     * channel通道活跃时(刚连接上注册中心)
     * 发送RPC请求给注册中心,请求感兴趣的服务集合以获取对应的提供者节点信息
     *
     * @param ctx channel信息上下文
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 获取RemoteInvoker单例,并添加提供者信息(对于消费者来说,注册中心的拉取服务是提供者服务)
        RemoteInvoker remoteInvoker = SingletonUtils.getInstance(RemoteInvoker.class);
        remoteInvoker.addProvider(hust.rpc.pojo.Channel.wrap(ctx.channel()), Collections.singletonList(Constants.CONSUMER_SERVICE));

        // 获取Proxy单例(AllProxy实现)
        Proxy proxy = SingletonUtils.getInstance(Proxy.class);
        // 获取ConsumerService代理对象
        ConsumerService service = proxy.getProxy(ConsumerService.class);

        // 通过连接池发送拉取请求
        ThreadUtils.execute(() -> {
                    // 获取扫描到的远程服务集合
                    List<String> remoteServices = proxy.getRemoteServices();
                    // 移除当前拉取服务
                    remoteServices.remove(Constants.CONSUMER_SERVICE);
                    log.info("消费者请求服务集合: {}", remoteServices);

                    // 通过代理对象发送RPC请求,获取远程服务对应的提供者节点信息
                    Map<String, List<Host>> serviceInfo = service.pullProviders(remoteServices);

                    // TODO 可以做改进
                    Map<Host, List<String>> newServiceInfo = new HashMap<>();
                    serviceInfo.forEach((s, v) -> {
                        for (Host host : v) {
                            List<String> services = newServiceInfo.getOrDefault(host, new ArrayList<>());
                            services.add(s);
                            newServiceInfo.put(host, services);
                        }
                    });

                    // 添加提供者信息(真正用户层面的服务对应的提供者信息)
                    remoteInvoker.addProviders(newServiceInfo);

                    // 闭锁打开,同步结束,监听本地端口
                    ConsumerBootstrap.countDown();
                }
        );
        super.channelActive(ctx);
    }

    @Override
    protected void handleResponse(Channel channel, RpcResponse response) {
        log.info(response.toString());
        super.handleResponse(channel, response);
    }
}
